package net.bwie.realtime.jtp.dwd.log.lx2.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class AppLogSplitProcessFunction extends ProcessFunction<String, String> {
    private final OutputTag<String> errorTag;
    private final OutputTag<String> startTag;
    private final OutputTag<String> actionsTag;
    private final OutputTag<String> displaysTag;

    public AppLogSplitProcessFunction(OutputTag<String> errorTag, OutputTag<String> startTag, OutputTag<String> actionsTag, OutputTag<String> displaysTag) {
        this.errorTag = errorTag;
        this.startTag = startTag;
        this.actionsTag = actionsTag;
        this.displaysTag = displaysTag;
    }



    @Override
    public void processElement(String s, ProcessFunction<String, String>.Context context,
                               Collector<String> collector) throws Exception {
        // 将数据转换成jsonobject 格式
       JSONObject jsonObject = JSON.parseObject(s);


        // 将错误日志提取出来
        if (null!= jsonObject.get("err")){
            context.output(errorTag,s);
        }

        // 启动日志
        if(null != jsonObject.getString("start")){
            context.output(startTag,jsonObject.toJSONString());
        }
        // 页面日志 其中包括 曝光日志和动作日志
        else{
            // 提取曝光和动作数据
            com.alibaba.fastjson.JSONArray displays = jsonObject.getJSONArray("displays");
            com.alibaba.fastjson.JSONArray actions = jsonObject.getJSONArray("actions");

            // 删除曝光和动作数据
            jsonObject.remove("displays");
            jsonObject.remove("actions");

            // 输出页面日志
            collector.collect(jsonObject.toJSONString());


            // 曝光日志
            if(null != displays && !displays.isEmpty()){
                for (int i = 0; i < displays.size(); i++) {
                    JSONObject jsonObject1 = displays.getJSONObject(i);

                    context.output(displaysTag,jsonObject1.toJSONString());
                }
            }
            // 动作日志
            if(null != actions && !actions.isEmpty()){
                for (int i = 0; i < actions.size(); i++) {
                    JSONObject jsonObject1 = actions.getJSONObject(i);

                    context.output(actionsTag,jsonObject1.toJSONString());
                }
            }

        }

    }
}